package com.gzyj.flink.gps.map;

import com.gzyj.flink.gps.GPSPosition;
import com.gzyj.flink.gps.addtionalfnfo.JTT808_AddtionalInfo;
import com.gzyj.flink.gps.addtionalfnfo.subiao.JTT808_0x0200_0x64;
import com.gzyj.flink.gps.addtionalfnfo.subiao.JTT808_0x0200_0x65;
import com.gzyj.flink.pojo.gps.TCmdannex;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.util.Collector;

import java.util.Date;

public class ZdaqFlow {

    private SingleOutputStreamOperator<GPSPosition> gpsSingleStream;

    public ZdaqFlow(SingleOutputStreamOperator<GPSPosition> gpsSingleStream) {
        this.gpsSingleStream = gpsSingleStream;
    }

    public void processZdaq() {

        gpsSingleStream.filter((gps) -> {
            return StringUtils.isNotEmpty(gps.getJtt808_0x65_platformId()) || StringUtils.isNotEmpty(gps.getJtt808_0x64_platformId());
        }).flatMap(new FlatMapFunction<GPSPosition, TCmdannex>() {
            @Override
            public void flatMap(GPSPosition gpsPosition, Collector<TCmdannex> collector) throws Exception {
                if (gpsPosition.getVersion().equals("2019")) {

                } else {

                    for (JTT808_AddtionalInfo addtionalInfo : gpsPosition.getGnssDataAdditionalInfo()) {

                        if (addtionalInfo.getID() == 0x64) {

                            JTT808_0x0200_0x64 jtt808_0x0200_0x64 = (JTT808_0x0200_0x64) addtionalInfo;


                            if (jtt808_0x0200_0x64.getAlarmIdentification().getAttachCount() > 0) {

                                TCmdannex tCmdannex = new TCmdannex() {{
                                    alarmNumber = gpsPosition.getJtt808_0x64_platformId();
                                    pickTime = new Date();
                                    fileState = 1;
                                }};

                                collector.collect(tCmdannex);

                            } else {
                                TCmdannex tCmdannex = new TCmdannex() {{
                                    alarmNumber = gpsPosition.getJtt808_0x64_platformId();
                                    pickTime = new Date();
                                    downTime = new Date();
                                    fileState = 4;
                                }};

                                collector.collect(tCmdannex);

                            }


                        } else if (addtionalInfo.getID() == 0x65) {

                            JTT808_0x0200_0x65 jtt808_0x0200_0x65 = (JTT808_0x0200_0x65) addtionalInfo;


                            if (jtt808_0x0200_0x65.getAlarmIdentification().getAttachCount() > 0) {

                                TCmdannex tCmdannex = new TCmdannex() {{
                                    alarmNumber = gpsPosition.getJtt808_0x65_platformId();
                                    pickTime = new Date();
                                    fileState = 1;
                                }};

                                collector.collect(tCmdannex);

                            } else {
                                TCmdannex tCmdannex = new TCmdannex() {{
                                    alarmNumber = gpsPosition.getJtt808_0x65_platformId();
                                    pickTime = new Date();
                                    downTime = new Date();
                                    fileState = 4;
                                }};

                                collector.collect(tCmdannex);

                            }

                        }


                    }


                }


            }
        }).addSink(new RichSinkFunction<TCmdannex>() {
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
            }

            @Override
            public void close() throws Exception {
                super.close();
            }

            @Override
            public void invoke(TCmdannex value, Context context) throws Exception {
                super.invoke(value, context);
            }
        });


    }


}
